package simpledb;

import java.util.*;
import java.io.*;

/**
 * Each instance of HeapPage stores data for one page of HeapFiles and 
 * implements the Page interface that is used by BufferPool.
 *
 * @see HeapFile
 * @see BufferPool
 *
 */
public class HeapPage implements Page {

    final HeapPageId pid;
    final TupleDesc td;
    //header表示元组对应的slot是否有效，1有效，0无效
    //假设header[0]=15  转换二进制00001111  注意slot从低位到高位对应
    //此时表示第0，1，2，3个slot是有效的
    final byte header[];
    final Tuple tuples[];
    //等同于tuple的个数
    final int numSlots;

    byte[] oldData;
    private final Byte oldDataLock=new Byte((byte)0);

    TransactionId transactionId;

    /**
     * Create a HeapPage from a set of bytes of data read from disk.
     * The format of a HeapPage is a set of header bytes indicating
     * the slots of the page that are in use, some number of tuple slots.
     *  Specifically, the number of tuples is equal to: <p>
     *          floor((BufferPool.getPageSize()*8) / (tuple size * 8 + 1))
     * <p> where tuple size is the size of tuples in this
     * database table, which can be determined via {@link Catalog#getTupleDesc}.
     * The number of 8-bit header words is equal to:
     * <p>
     *      ceiling(no. tuple slots / 8)
     * <p>
     * @see Database#getCatalog
     * @see Catalog#getTupleDesc
     * @see BufferPool#getPageSize()
     */
    public HeapPage(HeapPageId id, byte[] data) throws IOException {
        this.pid = id;
        this.td = Database.getCatalog().getTupleDesc(id.getTableId());
        this.numSlots = getNumTuples();
        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(data));

        // allocate and read the header slots of this page
        header = new byte[getHeaderSize()];
        for (int i=0; i<header.length; i++)
            header[i] = dis.readByte();
        
        tuples = new Tuple[numSlots];
        try{
            // allocate and read the actual records of this page
            for (int i=0; i<tuples.length; i++)
                tuples[i] = readNextTuple(dis,i);
        }catch(NoSuchElementException e){
            e.printStackTrace();
        }
        dis.close();

        setBeforeImage();
    }

    /** Retrieve the number of tuples on this page.
        @return the number of tuples on this page
    */
    private int getNumTuples() {        
        // some code goes here
        int tupsPerPage = (BufferPool.getPageSize() * 8) / ( td.getSize()* 8 + 1);
        return tupsPerPage;
    }

    /**
     * Computes the number of bytes in the header of a page in a HeapFile with each tuple occupying tupleSize bytes
     * @return the number of bytes in the header of a page in a HeapFile with each tuple occupying tupleSize bytes
     * 获得头部字节数
     */
    private int getHeaderSize() {        
        
        // some code goes here
        int headerSize = (getNumTuples()/8);
        if(getNumTuples()>headerSize*8)
        {
            headerSize++;
        }
        return headerSize;
    }
    
    /** Return a view of this page before it was modified
        -- used by recovery */
    public HeapPage getBeforeImage(){
        try {
            byte[] oldDataRef = null;
            synchronized(oldDataLock)
            {
                oldDataRef = oldData;
            }
            return new HeapPage(pid,oldDataRef);
        } catch (IOException e) {
            e.printStackTrace();
            //should never happen -- we parsed it OK before!
            System.exit(1);
        }
        return null;
    }
    
    public void setBeforeImage() {
        synchronized(oldDataLock)
        {
        oldData = getPageData().clone();
        }
    }

    /**
     * @return the PageId associated with this page.
     */
    public HeapPageId getId() {
    // some code goes here
        if(pid!=null)
          return pid;

        throw new UnsupportedOperationException("implement this");
    }

    /**
     * Suck up tuples from the source file.
     */
    private Tuple readNextTuple(DataInputStream dis, int slotId) throws NoSuchElementException {
        // if associated bit is not set, read forward to the next tuple, and
        // return null.
        if (!isSlotUsed(slotId)) {
            for (int i=0; i<td.getSize(); i++) {
                try {
                    dis.readByte();
                } catch (IOException e) {
                    throw new NoSuchElementException("error reading empty tuple");
                }
            }
            return null;
        }

        // read fields in the tuple
        Tuple t = new Tuple(td);
        RecordId rid = new RecordId(pid, slotId);
        t.setRecordId(rid);
        try {
            for (int j=0; j<td.numFields(); j++) {
                Field f = td.getFieldType(j).parse(dis);
                t.setField(j, f);
            }
        } catch (java.text.ParseException e) {
            e.printStackTrace();
            throw new NoSuchElementException("parsing error!");
        }

        return t;
    }

    /**
     * Generates a byte array representing the contents of this page.
     * Used to serialize this page to disk.
     * <p>
     * The invariant here is that it should be possible to pass the byte
     * array generated by getPageData to the HeapPage constructor and
     * have it produce an identical HeapPage object.
     *
     * @see #HeapPage
     * @return A byte array correspond to the bytes of this page.
     */
    public byte[] getPageData() {
        int len = BufferPool.getPageSize();
        ByteArrayOutputStream baos = new ByteArrayOutputStream(len);
        DataOutputStream dos = new DataOutputStream(baos);

        // create the header of the page
        for (int i=0; i<header.length; i++) {
            try {
                dos.writeByte(header[i]);
            } catch (IOException e) {
                // this really shouldn't happen
                e.printStackTrace();
            }
        }

        // create the tuples
        for (int i=0; i<tuples.length; i++) {

            // empty slot
            if (!isSlotUsed(i)) {
                for (int j=0; j<td.getSize(); j++) {
                    try {
                        dos.writeByte(0);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                continue;
            }

            // non-empty slot
            for (int j=0; j<td.numFields(); j++) {
                Field f = tuples[i].getField(j);
                try {
                    f.serialize(dos);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        // padding
        int zerolen = BufferPool.getPageSize() - (header.length + td.getSize() * tuples.length); //- numSlots * td.getSize();
        byte[] zeroes = new byte[zerolen];
        try {
            dos.write(zeroes, 0, zerolen);
        } catch (IOException e) {
            e.printStackTrace();
        }

        try {
            dos.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }

        return baos.toByteArray();
    }

    /**
     * Static method to generate a byte array corresponding to an empty
     * HeapPage.
     * Used to add new, empty pages to the file. Passing the results of
     * this method to the HeapPage constructor will create a HeapPage with
     * no valid tuples in it.
     *
     * @return The returned ByteArray.
     */
    public static byte[] createEmptyPageData() {
        int len = BufferPool.getPageSize();
        return new byte[len]; //all 0
    }

    /**
     * Delete the specified tuple from the page; the corresponding header bit should be updated to reflect
     *   that it is no longer stored on any page.
     *   删除一个元组
     *   判断是否在当前page中以及slot是否有效
     *   上述成立的话将tuples[i]赋空，并将当前页面标记为dirty
     *
     * @throws DbException if this tuple is not on this page, or tuple slot is
     *         already empty.
     * @param t The tuple to delete
     */
    public void deleteTuple(Tuple t) throws DbException {
        // some code goes here
        // not necessary for lab1
        RecordId recordId = t.getRecordId();
        PageId pageId = recordId.getPageId();
        if(!getId().equals(pageId))
        {
            throw new DbException("该元组不在当前page");
        }
        int tupleNumber = recordId.getTupleNumber();
        if(!isSlotUsed(tupleNumber))
        {
            throw new DbException("该元组不合法");
        }
        markSlotUsed(tupleNumber,false);
        tuples[tupleNumber] = null;
    }

    /**
     * Adds the specified tuple to the page;  the tuple should be updated to reflect
     *  that it is now stored on this page.
     *         插入一个元组
     *        判断是否在当前page中以及slot是否有效
     *        上述成立的话将tuples[i]=tuple，并将当前页面标记为dirty
     * @throws DbException if the page is full (no empty slots) or tupledesc
     *         is mismatch.
     * @param t The tuple to add.
     */
    public void insertTuple(Tuple t) throws DbException {
        // some code goes here
        // not necessary for lab1
        int numEmptySlots = getNumEmptySlots();
        if(numEmptySlots==0)
        {
            throw new DbException("page已满");
        }
        if(!td.equals(t.getTupleDesc()))
        {
            throw new DbException("类型不匹配");
        }
        int insert_pos = numSlots-numEmptySlots;
        RecordId recordId = t.getRecordId();
        if(null==recordId)
        {
            t.setRecordId(new RecordId(pid,insert_pos));
        }
        tuples[insert_pos] = t;
        markSlotUsed(insert_pos,true);
    }

    /**
     * Marks this page as dirty/not dirty and record that transaction
     * that did the dirtying
     * 标记当前页面为dirty
     */
    public void markDirty(boolean dirty, TransactionId tid) {
        // some code goes here
	    // not necessary for lab1
        if(dirty)
        {
            transactionId = tid;
        }
        else
        {
            transactionId = null;
        }
    }

    /**
     * Returns the tid of the transaction that last dirtied this page, or null if the page is not dirty
     */
    public TransactionId isDirty() {
        // some code goes here
	// Not necessary for lab1
        return transactionId;
    }

    /**
     * Returns the number of empty slots on this page.
     * 获得空/无效的slot个数
     */
    public int getNumEmptySlots() {
        // some code goes here
        int numEmptySlots = 0;
        for (int i = 0; i <numSlots;i++)
        {
            int slot_header_pos = i/8;
            //slot_header_pos 表示slot在byte[]中的索引位置
            int slot_byte_index = i%8;
            //slot_byte_index 表示slot在一个byte中对应0还是1
            if(!isSlotUsed(i))
            {
                numEmptySlots++;
            }
        }
        return numEmptySlots;
    }

    /**
     * Returns true if associated slot on this page is filled.
     * 判断元组对应的slot是否有效
     */
    public boolean isSlotUsed(int i) {
        // some code goes here
        //slot_header_pos 表示slot在header[]的下标
        int slot_header_pos = i/8;
        //slot_byte_index 表示slot在一个byte中对应0还是1
        int slot_byte_index = i%8;
        if(slot_header_pos>=header.length)
            throw new NoSuchElementException();
        byte value = header[slot_header_pos];
        //使用左移，如果对应的slot是1则为负数，对应的slot为0为正数（补码）
        return (byte)(value<<(7-slot_byte_index))<0;
    }

    /**
     * Abstraction to fill or clear a slot on this page.
     * i是tupleNumber,对应slot位置。
     * value为true时  将对应的slot改为1  表示合法 例如插入一个新元组
     * value为false时  将对应slot该为0  表示不合法  例如删除一个元组
     */
    private void markSlotUsed(int i, boolean value) {
        // some code goes here
        // not necessary for lab1
        if(i>getNumTuples())
            throw new NoSuchElementException();
        int slot_header_pos = i/8;
        //slot_header_pos 表示slot在byte[]中的索引位置
        int slot_byte_index = i%8;
        //slot_byte_index 表示slot在一个byte中对应0还是1
        byte slot_header = header[slot_header_pos];
        if(value)
        {
            header[slot_header_pos] = (byte) (slot_header+(1<<slot_byte_index));
        }
        else
        {
            header[slot_header_pos] = (byte) (slot_header-(1<<slot_byte_index));
        }
    }

    /**
     * @return an iterator over all tuples on this page (calling remove on this iterator throws an UnsupportedOperationException)
     * (note that this iterator shouldn't return tuples in empty slots!)
     */
    public Iterator<Tuple> iterator() {
        // some code goes here
        return new heapPageIterator();
    }

    //heapPageIterator迭代器的类，实现Iterator接口
    //实现hasNext()和next()两个方法
    private class  heapPageIterator implements Iterator<Tuple>
    {
        int cursor;
        int lastRet = -1;
        int size = numSlots;
        int usedNumTuples = 0;
        @Override
        public boolean hasNext() {
            if(cursor>=numSlots)
                return false;
            for(int i = cursor; i <numSlots;i++)
            {
                if(isSlotUsed(i))
                {
                    return true;
                }
            }
            return false;
        }

        @Override
        public Tuple next() {

            while(hasNext()&&!isSlotUsed(cursor))
            {
                cursor++;
            }
            if(cursor<numSlots)
            {
                return tuples[cursor++];
            }
            else
                return null;
        }

        ;
    }

}

